#include "Eventloop.h"
#include "EventloopProxy.h"
#include "IOchannel.h"
#include "log/Logger.h"
#include "thread/current_thread.h"
#include "timer/TimerManager.h"
#include "utility/assertion.h"
#include "wrap/eventfd.h"
#include "wrap/unistd.h"



namespace _ {

using WakeType = uint64_t;

static int create_wake_fd() {
    const auto fd = Wrap::eventfd_(0, Wrap::efd_nonblock() | Wrap::efd_cloexec());
    if (fd < 0) {
        LOG_SYSERROR << "Fail to create wake fd";
        ::abort();
    }
    return fd;
}

static void wake_event_func(int fd) {
    WakeType one = 1;
    const auto n = Wrap::read_(fd, &one, sizeof(WakeType));
    if (n == sizeof(WakeType)) {
        return;
    }
    LOG_ERROR << "weak reads "  << n << " bytes instead of " << sizeof(WakeType);
} 

static void wake_up(int fd) {
    static constexpr WakeType ONE = 1;
    const auto n = Wrap::write_(fd, &ONE, sizeof(WakeType));
    if (n == sizeof(WakeType)) {
        return ;
    }
    LOG_ERROR << "wake writes " << n << " bytes instead of " << sizeof(WakeType);
}


static thread_local Eventloop *loopOfCurrentThread = nullptr;

}


Eventloop *Eventloop::LOOP_OF_CURRENT_THREAD() {
    return _::loopOfCurrentThread;
}

Eventloop::Eventloop():
    _mutex{},
    _pendingFuncs{},
    _eventloopProxy{},
    _timerManager{},
    _wakeChannel{},
    _tid{CurrentThread::tid()},
    _callingPendingFuncs{false},
    _looping{false},
    _loopOver{false},
    _quit{false}
{
    LOG_DEBUG << "Eventloop " << this << "created in thread" << _tid;
    if (_::loopOfCurrentThread != nullptr) {
        LOG_FATAL << "Another loop " << _::loopOfCurrentThread
                  << " existed in this thread " << _tid;
    }
    else {
        _::loopOfCurrentThread = this;
    }
    _eventloopProxy = std::make_unique<EventloopProxy>(*this);
    _timerManager = std::make_unique<TimerManager>(*this);
    _wakeChannel = std::make_unique<IOchannel>(
                    *this, _::create_wake_fd(),
                    std::string{"wake"} + CurrentThread::tid_string());
    _wakeChannel->set_read_callback(
        [fd = _wakeChannel->fd()](Utility::timepoint /*t*/){
            _::wake_event_func(fd);
        });
    _wakeChannel->enable_reading();
}

Eventloop::~Eventloop() {
    LOG_DEBUG << "Eventloop " << this << " of this thread" << _tid
              << " destructs in thread " << CurrentThread::tid();
    _wakeChannel->disable();
    _wakeChannel->remove();
    Wrap::close_(_wakeChannel->fd());
    _::loopOfCurrentThread = nullptr;
}

void Eventloop::loop() {
    assert(!_looping);
    ASSERT_IN_THIS_THREAD();
    _looping = true;
    _loopOver = false;
    _quit = false;
    LOG_VERBOSE << "Eventloop " << this << " start looping";
    while (!_quit) {
        _eventloopProxy->poll_event();
        _call_pending_funcs();
    }
    LOG_VERBOSE << "Eventloop " << this << "stop looping";
    _call_pending_funcs();
    _looping = false;
    _loopOver = true;
}

void Eventloop::assert_in_this_thread(const char *func, const char *file, int line) const {
    if (is_in_this_thread()) {
        return;
    }
    LOG_FATAL << "Eventloop " << this << " was created in " << _tid
              << ", current tid=" << CurrentThread::tid() << " in " << func
              << "() of " << file << ":" << line;
}

bool Eventloop::is_in_this_thread() const {
    return _tid == CurrentThread::tid();
}

void Eventloop::quit() {
    assert(!_quit);
    _quit = true;
    if (is_in_this_thread()) {
        return;
    }
    _::wake_up(_wakeChannel->fd());
}

void Eventloop::run_in_loop(const TrivialFunc &func) {
    if (!is_in_this_thread()) {
        queue_in_loop(func);
        return;
    }
    assert(func);
    func();
}

void Eventloop::queue_in_loop(const TrivialFunc &func) {
    if (_loopOver) {
        return;
    }
    assert(func);
    {
        const std::lock_guard guard{_mutex};
        _pendingFuncs.emplace_back(func);
    }
    if (is_in_this_thread() && !_callingPendingFuncs) {
        return;
    }
    _::wake_up(_wakeChannel->fd());
}


void Eventloop::cancel(const TimerID &timerid) {
    _timerManager->cancel(timerid);
}

TimerID Eventloop::run_after(double delay, const TrivialFunc &callback) {
    return _timerManager->add(callback, Utility::cast_duration(delay), true);
}

TimerID Eventloop::run_every(double delay, const TrivialFunc &callback) {
    return _timerManager->add(callback, Utility::cast_duration(delay), false);
}

void Eventloop::_call_pending_funcs() {
    std::vector<TrivialFunc> funcs{};
    _callingPendingFuncs = true;
    {
        const std::lock_guard guard{_mutex};
        funcs.swap(_pendingFuncs);
    }
    for (const auto &x : funcs) {
        x();
    }
    _callingPendingFuncs = false;
}